package com.rrd.hive.udf;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;

public class Four_args_MappingUDF extends UDF {
    Text text = new Text();
    String sys_mapping_code = "";
    int fiel_type2 = 0;
    List dw_code_maping = new ArrayList();
    public Text evaluate(String code_type,String table_name,String src_code,String src_code2) throws Exception{

        String result = null;
        if(fiel_type2 == 0) {
            String code_type_get = "/data/cdm/" + code_type + "/000000_0";
            Configuration conf = new Configuration();
            /*FileSystem fs = FileSystem
                    .newInstance(URI.create(code_type_get), conf);
            fs.copyToLocalFile(false, new Path(code_type_get),
                    new Path("/tmp/udf/" + code_type + "/" + code_type + ".txt"));*/
            //String mapping_table_path = "/tmp/udf/" + code_type + "/" + code_type + ".txt";
            String filePath = "/data/cdm/t07_minor_sys_source_pa_all/000000_0";
            /*FileSystem fs2 = FileSystem
                    .newInstance(URI.create(filePath), conf);
            fs2.copyToLocalFile(false, new Path(filePath),
                    new Path("/tmp/udf/t07_minor_sys_source_pa_all/t07_minor_sys_source_pa_all.txt"));*/
           // String sys_table_path = "/tmp/udf/t07_minor_sys_source_pa_all/t07_minor_sys_source_pa_all.txt";
        /*String mapping_table_path = "E:\\2019工作\\数据仓库模型\\ddl\\编码\\mapping\\t07_creditor_mapping_pa_all\\000000_0";
        String sys_table_path = "E:\\2019工作\\数据仓库模型\\ddl\\编码\\mapping\\sys_000000_0";*/
            String sys_code = StringUtils.substringBefore(table_name, "_");
            HashMap<String, String> maping = readTxtFile(filePath,conf);
            sys_mapping_code = maping.get(sys_code);
            if(sys_mapping_code == null){
                throw new Exception("未获得源系统映射值");
            }
            List<String> type_one = new ArrayList<>(Arrays.asList("t07_creditor_mapping_pa_all", "t07_repay_subject_mapping_pa_all"));
            List<String> type_two = new ArrayList<>(Arrays.asList("t07_repay_status_mapping_pa_all"));
            dw_code_maping = readcodeFile(code_type_get,conf);
            if (type_one.contains(code_type)) {
                fiel_type2 = 2;
            } else if (type_two.contains(code_type)) {
                fiel_type2 = 3;
            } else {
                fiel_type2 = 4;
            }
        }
        String[] fieldValue = null;
        if(fiel_type2==2&&src_code2.equals("1")) {

            for(int i =0 ;i<dw_code_maping.size();i++) {
                Object line = dw_code_maping.get(i);
                String lineTxt = line.toString();
                fieldValue = lineTxt.split(",");
                //System.out.println("fieldValue:"+fieldValue);
                if(fieldValue[7].equals("1")) {
                    if(fieldValue[3].equals(sys_mapping_code)&&fieldValue[4].equals(src_code)){
                        result=fieldValue[0];
                        break;
                    }
                    //将读取的每一行的记录存入到list集合中
                }
            }
        }else if(fiel_type2==2&&src_code2.equals("2")) {

            for(int i =0 ;i<dw_code_maping.size();i++) {
                Object line = dw_code_maping.get(i);
                String lineTxt = line.toString();
                fieldValue = lineTxt.split(",");
                //System.out.println("fieldValue:"+fieldValue);
                if(fieldValue[8].equals("1")) {
                    if(fieldValue[3].equals(sys_mapping_code)&&fieldValue[4].equals(src_code)){
                        result=fieldValue[0];
                        break;
                    }
                    //将读取的每一行的记录存入到list集合中
                }
            }
        }else if(fiel_type2==3) {
            for(int i =0 ;i<dw_code_maping.size();i++) {
                Object line = dw_code_maping.get(i);
                String lineTxt = line.toString();
                fieldValue = lineTxt.split(",");
                //System.out.println("fieldValue:"+fieldValue);
                if(fieldValue[3].equals(sys_mapping_code)&& fieldValue[4].equals(src_code)&&fieldValue[7].equals(src_code2)) {
                        result=fieldValue[0];
                        break;
                    //将读取的每一行的记录存入到list集合中
                }
            }
        }else{
            System.out.println("不存在的代码类型");
        }
        if(result == null){
            result = "99";
        }
        text.set(result);
        /*File mapping_fiel =new File(mapping_table_path);
        mapping_fiel.delete();
        File sys_table_fiel =new File(sys_table_path);
        sys_table_fiel.delete();*/
        return text;
    }



    public static HashMap<String, String> readTxtFile(String filePath,Configuration conf) throws Exception {
        HashMap<String, String> map =  new HashMap<String, String>();

        try {
            FileSystem fs = FileSystem
                    .newInstance(URI.create(filePath), conf);
        Path remotePath = new Path(filePath);
        FSDataInputStream in = fs.open(remotePath);
        if (fs.exists(remotePath)) {
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(in));
        String lineTxt = null;
         int count = 0;
         String[] fieldValue = null;
         while ((lineTxt = bufferedReader.readLine()) != null) {
                fieldValue = lineTxt.split(",");
                //System.out.println("fieldValue:"+fieldValue);
                map.put(fieldValue[8], fieldValue[0]);
                //将读取的每一行的记录存入到list集合中
                count++;
            }
            bufferedReader.close();
            in.close();
            fs.close();
        }
            else {
            System.out.println("找不到指定的文件");
            throw new Exception("找不到指定的文件");
        }
        } catch (Exception e) {
            System.out.println("读取文件内容出错");
            e.printStackTrace();
            throw new Exception("读取文件内容出错");
        }
        return map;
    }


    public static List readcodeFile(String filePath,Configuration conf) throws Exception {
        List list = new ArrayList();
        try {
            FileSystem fs = FileSystem
                    .newInstance(URI.create(filePath), conf);
            Path remotePath = new Path(filePath);
            FSDataInputStream in = fs.open(remotePath);
            if (fs.exists(remotePath)) {
                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(in));
                String lineTxt = null;
                int count = 0;
                //定义每条记录取出的字段值数组
                String[] fieldValue = null;
                //定义Map集
                while ((lineTxt = bufferedReader.readLine()) != null) {
                    list.add(lineTxt);
                    count++;
                }
                bufferedReader.close();
                in.close();
                fs.close();
            } else {
                System.out.println("找不到指定的文件");
                throw new Exception("找不到指定的文件");
            }
        }catch (Exception e) {
            System.out.println("读取文件内容出错");
            e.printStackTrace();
            throw new Exception("读取文件内容出错");
        }
        return list;
    }


    /*public static HashMap<String, String> readTxtFile2(String filePath){
        //定义存储读取到的数据记录的集合
        HashMap<String, String> map =  new HashMap<String, String>();
        try {
            String encoding="UTF-8";
            File file=new File(filePath);
            //判断文件是否存在
            if(file.isFile() && file.exists()){
                //考虑到编码格式
                InputStreamReader read = new InputStreamReader(new FileInputStream(file),encoding);
                BufferedReader bufferedReader = new BufferedReader(read);
                String lineTxt = null;
                //记录读取的数据文件的行数
                int count = 0;
                //定义每条记录取出的字段值数组
                String[] fieldValue = null;
                //定义Map集
                while ((lineTxt = bufferedReader.readLine()) != null) {
                    fieldValue = lineTxt.split(",");
                    //System.out.println("fieldValue:"+fieldValue);
                    map.put(fieldValue[8], fieldValue[0]);
                    //将读取的每一行的记录存入到list集合中
                    count++;
                }
                read.close();
            }else{
                System.out.println("找不到指定的文件");
            }
        } catch (Exception e) {
            System.out.println("读取文件内容出错");
            e.printStackTrace();
        }
        return map;
    }*/

   /* public static List readcodeFile2(String filePath){
        //定义存储读取到的数据记录的集合
        List list = new ArrayList();
        try {
            String encoding="UTF-8";
            File file=new File(filePath);
            //判断文件是否存在
            if(file.isFile() && file.exists()){
                //考虑到编码格式
                InputStreamReader read = new InputStreamReader(new FileInputStream(file),encoding);
                BufferedReader bufferedReader = new BufferedReader(read);
                String lineTxt = null;
                //记录读取的数据文件的行数
                int count = 0;
                //定义每条记录取出的字段值数组
                String[] fieldValue = null;
                //定义Map集
                while ((lineTxt = bufferedReader.readLine()) != null) {
                    list.add(lineTxt);
                    count++;
                }
                read.close();
            }else{
                System.out.println("找不到指定的文件");
            }
        } catch (Exception e) {
            System.out.println("读取文件内容出错");
            e.printStackTrace();
        }
        return list;
    }*/




   /* public static void main(String[] args) throws Exception {
        String args1 = "t07_creditor_mapping_pa_all";
        String args2 = "uni01_loan_subject_info_chain";
        String args3 = "PERSONAL";
        String args4 = "1";
        Four_args_MappingUDF dfu = new Four_args_MappingUDF();
        System.out.println(dfu.evaluate(args1,args2,args3,args4));
    }*/
}
